package cn.edu.flink.scala.tutorial.state.keyedstate

import org.apache.flink.api.common.functions.RichMapFunction
import org.apache.flink.api.common.state.{MapState, MapStateDescriptor}
import org.apache.flink.configuration.Configuration
import org.apache.flink.streaming.api.scala._

object MapStateTest {
  def main(args: Array[String]): Unit = {
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    env.setParallelism(1)

    val range1 = (1 to 15).toList
    val range2 = (15 to 10 by -1).toList
    val value = env.fromCollection(range1 ::: range2)

    value
      .keyBy(x => true)
      .map(new MyMap)
      .print()

    env.execute("值状态")
  }


  class MyMap extends RichMapFunction[Int, (String, Int)] {
    // 定义值状态
    var mapState: MapState[Int, Int] = _


    override def open(parameters: Configuration): Unit = {
      mapState = getRuntimeContext.getMapState(new MapStateDescriptor[Int, Int]("map-state", classOf[Int], classOf[Int]))
    }

    override def map(value: Int): (String, Int) = {
      val i = if (mapState.contains(value)) mapState.get(value) else 0
      mapState.put(value, i+1)

      (value.toString, mapState.get(value))
    }
  }
}
